package com.atguigu.day07;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

// select * from A inner join B on A.key = B.key;
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<ClickEvent> clickSource1 = env.addSource(new ClickSource());

        DataStreamSource<ClickEvent> clickSource2 = env.addSource(new ClickSource());

        clickSource1.keyBy(r -> r.username)
                .connect(clickSource2.keyBy(r -> r.username))
                .process(new InnerJoin())
                .print();

        env.execute();
    }

    public static class InnerJoin extends CoProcessFunction<ClickEvent, ClickEvent, String> {
        private ListState<ClickEvent> history1;
        private ListState<ClickEvent> history2;

        @Override
        public void open(Configuration parameters) throws Exception {
            history1 = getRuntimeContext().getListState(new ListStateDescriptor<ClickEvent>("history1", Types.POJO(ClickEvent.class)));
            history2 = getRuntimeContext().getListState(new ListStateDescriptor<ClickEvent>("history2", Types.POJO(ClickEvent.class)));
        }

        @Override
        public void processElement1(ClickEvent in1, Context ctx, Collector<String> out) throws Exception {
            history1.add(in1);

            for (ClickEvent e : history2.get()) {
                out.collect(in1 + " => " + e);
            }
        }

        @Override
        public void processElement2(ClickEvent in2, Context ctx, Collector<String> out) throws Exception {
            history2.add(in2);

            for (ClickEvent e : history1.get()) {
                out.collect(e + " => " + in2);
            }
        }
    }
}
